package org.sg20.graphviewer;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.sg20.utils.Edge;
import org.sg20.utils.ParseText;

import java.io.IOException;

public class GraphViewerMapper extends Mapper<Text, Text, Text, DoubleWritable> {

    Text outKey;
    DoubleWritable outVal;

    @Override
    protected void setup(Mapper<Text, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
        super.setup(context);
        this.outKey=new Text();
        this.outVal=new DoubleWritable();
    }

    @Override
    protected void map(Text key, Text value, Mapper<Text, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
        Edge[] edges= ParseText.parseEdges(value.toString());
        String src,dest;
        for(Edge edge : edges){
            if(key.toString().compareTo(edge.getDestName())<0){
                src=key.toString();
                dest=edge.getDestName();
            }else{
                src=edge.getDestName();
                dest=key.toString();
            }
            outKey.set(src+","+dest);
            outVal.set(edge.getWeight());
            context.write(outKey,outVal);
        }
    }
}
